package com.qupeng.demo.kafka.kafkaspringbootconsumer.mm2.client;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class MM2ClientConsumerConfig<K, V> {
    @Autowired
    private MM2ClientProperties mm2ClientProperties;

    @Bean
    public Map<String, Object> mm2ClientConsumerConfigs() throws ClassNotFoundException {
        Map<String, Object> props = new HashMap<>();
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, mm2ClientProperties.getBootstrapServer());
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Class.forName(mm2ClientProperties.getKeyDeserializerInt()));
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(mm2ClientProperties.getValueDeserializerStr()));
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, mm2ClientProperties.getSecurityProtocol());
        if ("SSL".equals(mm2ClientProperties.getSecurityProtocol())) {
            props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, mm2ClientProperties.getSslKeystoreLocation());
            props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, mm2ClientProperties.getSslKeystorePassword());
            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, mm2ClientProperties.getSslTruststoreLocation());
            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, mm2ClientProperties.getSslTruststorePassword());
        }
        switch (mm2ClientProperties.getSerializationFormat()) {
            case "json":
                props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(mm2ClientProperties.getValueDeserializerJson()));
                break;
            case "avro":
                props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(mm2ClientProperties.getValueDeserializerAvro()));
                props.put("schema.registry.url", mm2ClientProperties.getSchemaRegistryUrl());
                break;
            default:
                props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Class.forName(mm2ClientProperties.getValueDeserializerStr()));
        }
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.CLIENT_ID_CONFIG, "mm2-client");
        return props;
    }
}
